"""异步工具函数，提供统一的异步操作处理机制"""

import asyncio
import inspect
import threading
from typing import Any, Callable, TypeVar, Optional, Dict, List
from concurrent.futures import ThreadPoolExecutor, Future
from loguru import logger

T = TypeVar('T')

# 线程池配置
DEFAULT_EXECUTOR = ThreadPoolExecutor(
    max_workers=10,
    thread_name_prefix="async_utils_thread"
)

# 存储每个线程的事件循环缓存
_event_loop_cache: Dict[int, asyncio.AbstractEventLoop] = {}


def get_or_create_event_loop() -> asyncio.AbstractEventLoop:
    """
    获取或创建事件循环，为每个线程维护独立的事件循环
    """
    thread_id = threading.get_ident()
    
    # 检查线程本地缓存中是否已有事件循环
    if thread_id in _event_loop_cache:
        loop = _event_loop_cache[thread_id]
        if loop.is_closed():
            del _event_loop_cache[thread_id]
        else:
            return loop
    
    # 创建新的事件循环并缓存
    try:
        # 尝试获取当前线程的事件循环（如果存在）
        loop = asyncio.get_event_loop()
    except RuntimeError:
        # 没有事件循环，创建新的
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
    
    # 缓存事件循环
    _event_loop_cache[thread_id] = loop
    return loop


async def run_sync_in_background(func: Callable[..., T], *args, **kwargs) -> T:
    """
    在后台线程中运行同步函数，不阻塞当前事件循环
    """
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(
        DEFAULT_EXECUTOR,
        lambda: func(*args, **kwargs)
    )


def run_async_in_sync(async_func: Callable[..., T], *args, **kwargs) -> T:
    """
    在同步上下文中运行异步函数
    智能处理不同的事件循环场景，避免阻塞和资源浪费
    """
    # 检查函数是否为协程函数
    if not inspect.iscoroutinefunction(async_func):
        # 如果不是异步函数，直接调用
        return async_func(*args, **kwargs)
    
    try:
        # 尝试获取当前事件循环
        loop = asyncio.get_event_loop()
        
        if loop.is_running():
            # 如果事件循环正在运行，使用线程池创建新的事件循环
            # 这样可以避免在运行中的事件循环上调用run_until_complete
            result = [None]
            error = [None]
            
            def thread_runner():
                try:
                    inner_loop = asyncio.new_event_loop()
                    asyncio.set_event_loop(inner_loop)
                    result[0] = inner_loop.run_until_complete(
                        async_func(*args, **kwargs)
                    )
                except Exception as e:
                    error[0] = e
                finally:
                    inner_loop.close()
            
            # 使用线程池执行，避免频繁创建和销毁线程
            future = DEFAULT_EXECUTOR.submit(thread_runner)
            try:
                # 等待结果，设置超时
                future.result(timeout=30)  # 30秒超时
            except TimeoutError:
                future.cancel()
                raise TimeoutError(f"异步函数执行超时: {async_func.__name__}")
            
            if error[0]:
                raise error[0]
            return result[0]
        else:
            # 事件循环存在但未运行，直接运行
            return loop.run_until_complete(async_func(*args, **kwargs))
    except RuntimeError:
        # 没有事件循环，创建新的
        return asyncio.run(async_func(*args, **kwargs))


class AsyncResult(Future):
    """异步结果包装器，提供同步接口访问异步结果"""
    
    def __init__(self, coro):
        super().__init__()
        self._coro = coro
        
    def result(self, timeout=None):
        if not self.done():
            # 如果结果未准备好，在当前线程中运行协程
            try:
                result = run_async_in_sync(self._coro)
                self.set_result(result)
            except Exception as e:
                self.set_exception(e)
        return super().result(timeout)


async def gather_with_timeout(*coros, timeout=30, return_exceptions=False) -> List[Any]:
    """
    带有超时的gather函数
    """
    try:
        return await asyncio.wait_for(
            asyncio.gather(*coros, return_exceptions=return_exceptions),
            timeout=timeout
        )
    except asyncio.TimeoutError:
        logger.error(f"gather操作超时: {timeout}秒")
        raise


async def safe_async_call(func: Callable[..., T], *args, **kwargs) -> T:
    """
    安全的异步函数调用，包含错误处理和超时保护
    """
    try:
        # 检查是否为协程函数
        if inspect.iscoroutinefunction(func):
            # 为异步函数添加超时保护
            return await asyncio.wait_for(
                func(*args, **kwargs),
                timeout=getattr(func, '__timeout__', 30)
            )
        else:
            # 同步函数在后台线程执行
            return await run_sync_in_background(func, *args, **kwargs)
    except asyncio.TimeoutError:
        logger.error(f"函数执行超时: {func.__name__}")
        raise
    except Exception as e:
        logger.error(f"函数执行异常: {func.__name__}, {str(e)}")
        raise


def async_timeout(seconds: float):
    """
    异步函数超时装饰器
    """
    def decorator(func):
        if not inspect.iscoroutinefunction(func):
            raise TypeError("@async_timeout只能用于异步函数")
        
        func.__timeout__ = seconds
        return func
    return decorator


# 清理函数，在应用关闭时调用
def cleanup_event_loops():
    """
    清理所有缓存的事件循环
    """
    for thread_id, loop in list(_event_loop_cache.items()):
        if not loop.is_closed():
            try:
                # 取消所有任务
                for task in asyncio.all_tasks(loop=loop):
                    task.cancel()
                # 关闭事件循环
                loop.run_until_complete(loop.shutdown_asyncgens())
                loop.close()
            except Exception as e:
                logger.error(f"关闭事件循环失败: {e}")
        del _event_loop_cache[thread_id]
    
    # 关闭线程池
    DEFAULT_EXECUTOR.shutdown(wait=True)
    logger.info("异步工具清理完成")
